package wnco.aggr;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import wnco.base.SomData;

public class AggreFunctionW {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<SomData> dss = env.fromElements(
                new SomData("cc", 6, 358L),
                new SomData("bb", 4, 745451L),
                new SomData("bb", 5, 51L),
                new SomData("aa", 1, 496656L),
                new SomData("aa", 2, 4956L),
                new SomData("aa", 3, 49568L),
                new SomData("bb", 4, 745451L),
                new SomData("bb", 5, 51L)

        );
//        KeyedStream<SomData, String> keyBy = dss.keyBy(new KeySelector<SomData, String>() {
//            @Override
//            public String getKey(SomData value) throws Exception {
//                return value.getName();
//            }
//        });
//        获取最大各分区最大值
//        keyBy.max("ctime").print("max: ");

        SingleOutputStreamOperator<Tuple2<String, Long>> map = dss.map(new MapFunction<SomData, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(SomData value) throws Exception {
                return Tuple2.of(value.getName(),1L);
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> reduce2 = map
                .keyBy(data -> data.f0)
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> reduce3 = reduce2.keyBy(data -> "all").reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                return value1.f1 > value2.f1 ? value1 : value2;
            }
        });
        reduce3.print();
        env.execute();
    }
}

